弹力设计篇之"限流设计" - [2026重制版]

resilience4j:
  ratelimiter:
    instances:
      # 核心API的全局限流（严格）
      coreApi:
        limitForPeriod: 100              # 每周期允许100个请求
        limitRefreshPeriod: 1s            # 刷新周期：每秒
        timeoutDuration: 0                # 等待时间：0表示不等待直接拒绝
        registerHealthIndicator: true     # 注册健康检查
        
      # 支付接口的特殊限流（更严格）
      paymentApi:
        limitForPeriod: 20               # 每秒最多20次支付请求
        limitRefreshPeriod: 1s
        timeoutDuration: 100ms            # 允许等待100ms获取令牌
        # 注意: timeoutDuration > 0 时会等待而非立即拒绝
        
      # 外部API调用的宽松限流
      externalApi:
        limitForPeriod: 500              # 每分钟500次
        limitRefreshPeriod: 60s          # 分钟级限流
        timeoutDuration: 0
        
      # 用户维度的动态限流（需自定义实现）
      userBasedApi:
        limitForPeriod: 10               # 每用户每分钟10次
        limitRefreshPeriod: 60s
        timeoutDuration: 0

@Service
@RequiredArgsConstructor
@Slf4j
public class RateLimitedService {
 
    private final ExternalApiClient apiClient;
    private final RateLimiterRegistry rateLimiterRegistry;
 
    /**
     * 注解方式：简单易用
     */
    @RateLimiter(name = "coreApi", fallbackMethod = "fallbackGetData")
    public DataResponse getData(String id) {
        log.info("Processing request for id: {} (rate limited)", id);
        return apiClient.fetchData(id);
    }
 
    /**
     * 限流触发时的降级方法
     */
    public DataResponse fallbackGetData(String id, RequestNotPermittedException e) {
        log.warn("Rate limit exceeded for id={}", id);
        
        # 返回降级数据或提示用户稍后重试
        return DataResponse.rateLimited(
            "Too many requests, please try again later",
            429  # HTTP 429 Status Code
        );
    }
 
    /**
     * 编程式方式：更灵活的控制
     */
    public PaymentResult processPayment(PaymentRequest request) {
        RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("paymentApi");
 
        # 装饰Supplier
        Supplier<PaymentResult> supplier = RateLimiter.decorateSupplier(rateLimiter, () -> {
            log.info("Payment request permitted (rate limited)");
            return apiClient.charge(request);
        });
 
        try:
            return supplier.get();
        } catch (RequestNotPermittedException e):
            # 被限流了
            log.warn("Payment rate limit exceeded! WaitTime={}ms",
                     e.getWaitDuration().toMillis());
                     
            # 可选：返回具体还需要等待多久
            long waitMs = e.getWaitDuration().toMillis();
            
            return PaymentResult.rateLimited(
                String.format("Payment service busy, please wait %dms", waitMs),
                waitMs
            );
    }
}

/**
 * 基于Redis的滑动窗口限流器
 * 
 * 使用Lua脚本保证原子性
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class DistributedRateLimiter {
 
    private final StringRedisTemplate redisTemplate;
    
    /** Lua脚本：原子性的限流检查和计数 */
    private static final String RATE_LIMIT_SCRIPT =
        "local key = KEYS[1] " +
        "local limit = tonumber(ARGV[1]) " +
        "local window = tonumber(ARGV[2]) " +
        "local now = tonumber(ARGV[3]) " +
        "" +
        "-- 清除过期的窗口数据 " +
        "redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window) " +
        "" +
        "-- 获取当前窗口内的请求数 " +
        "local current = redis.call('ZCARD', key) " +
        "" +
        "if current < limit then " +
        "   -- 未超过限制，计数并放行 " +
        "   redis.call('ZADD', key, now, now .. ':' .. math.random()) " +
        "   redis.call('EXPIRE', key, window) " +
        "   return {1, limit - current - 1} " +  -- 返回{允许, 剩余额度}
        "else " +
        "   -- 超过限制，拒绝 " +
        "   local ttl = redis.call('PTTL', key) " +
        "   return {0, ttl} " +  -- 返回{拒绝, 重试等待时间(ms)}
        "end";
 
    /**
     * 检查是否允许通过（滑动窗口算法）
     *
     * @param key 限流键（如 "ratelimit:user:123:api:order"）
     * @param limit 时间窗口内的最大请求数
     * @param windowSeconds 窗口大小（秒）
     * @return RateLimitResult 包含是否允许和剩余信息
     */
    public RateLimitResult tryAcquire(String key, int limit, int windowSeconds) {
        long now = System.currentTimeMillis();
        
        DefaultRedisScript<List<Long>> script = new DefaultRedisScript<>();
        script.setScriptText(RATE_LIMIT_SCRIPT);
        script.setResultType(List.class);
 
        // 执行Lua脚本
        List<Long> result = redisTemplate.execute(
            script,
            Collections.singletonList(getKey(key)),  # KEYS[1]
            String.valueOf(limit),                  // ARGV[1]: 限制数
            String.valueOf(windowSeconds * 1000),    // ARGV[2]: 窗口(毫秒)
            String.valueOf(now)                      // ARGV[3]: 当前时间戳
        );
 
        boolean allowed = result.get(0) == 1;
        long remaining = result.get(1);
 
        if (allowed) {
            log.debug("Rate limit check passed: key={}, remaining={}", key, remaining);
        } else {
            log.warn("Rate limit exceeded: key={}, retryAfter={}ms", key, remaining);
        }
 
        return RateLimitResult.builder()
            .allowed(allowed)
            .remaining(remaining)
            .key(key)
            .limit(limit)
            .windowSeconds(windowSeconds)
            .build();
    }
 
    private String getKey(String key) {
        return "ratelimit:" + key;
    }
}
 
/**
 * 使用示例：用户维度限流
 */
@RestController
@RequestMapping("/api/orders")
public class OrderController {
 
    private final DistributedRateLimiter rateLimiter;
 
    @PostMapping
    public ResponseEntity<?> createOrder(
            @RequestBody CreateOrderRequest request,
            @RequestHeader("X-User-ID") String userId) {
 
        // 构造限流Key：用户ID + API路径
        String limitKey = String.format("user:%s:api:POST:/orders", userId);
        
        // 检查限流：每用户每分钟最多10次下单
        RateLimitResult result = rateLimiter.tryAcquire(limitKey, 10, 60);
 
        if (!result.isAllowed()) {
            # 返回429 + Retry-After头
            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                .header("X-RateLimit-Limit", "10")
                .header("X-RateLimit-Remaining", "0")
                .header("Retry-After", String.valueOf(result.getRetryAfterMs() / 1000))
                .body(ErrorResponse.rateLimited("Too many requests"));
        }
 
        # 正常处理业务逻辑
        OrderResponse order = orderService.createOrder(request);
        
        return ResponseEntity.ok()
            .header("X-RateLimit-Limit", "10")
            .header("X-RateLimit-Remaining", String.valueOf(result.getRemaining()))
            .body(order);
    }
}

# istio-rate-limit.yaml
# 使用Envoy的Local Rate Limiting（全局限流）
apiVersion: networking.istio.io/v1beta1
kind: EnvoyFilter
metadata:
  name: order-service-ratelimit
  namespace: istio-system
spec:
  workloadSelector:
    labels:
      app: order-service
  configPatches:
  - applyTo: HTTP_FILTER
    match:
      context: SIDECAR_INBOUND
      listener:
        portNumber: 8080
        filterChain:
          filter:
            name: "envoy.filters.network.http_connection_manager"
            subType: 
    patch:
      operation: MERGE
      value:
        name: envoy.local_rate_limit_filter
        typed_config:
          "@type": type.http.filters.LocalRateLimit
          stat_prefix: http_local_rate_limiter
          token_bucket:
            max_tokens: 100           # 令牌桶容量
            tokens_filled_per_second: 100  # 每秒填充100个令牌
          filter_enabled:
            runtime_key: local_rate_limit_enabled
            default_value:
              numerator: 100
              denominator: 100
          response_status_code: 429    # 被限流时返回的状态码
          descriptors:
          - entries:
            - key: generic_key
              value: "order_service_limit"  # 通用限流标识
            
---
# 使用Envoy的Global Rate Limiting（需要外部Rate Limit Server）
# 更适合复杂的限流场景（如多维度：IP+User+API）
apiVersion: networking.istio.io/v1beta1
kind: EnvoyFilter
metadata:
  name: global-ratelimit
  namespace: istio-system
spec:
  configPatches:
  - applyTo: NETWORK_FILTER
    match:
      listener:
        portNumber: 8080
        filterChain:
          filter:
            name: "envoy.filters.network.http_connection_manager"
    patch:
      operation: MERGE
      value:
        name: envoy.filters.http.ratelimit
        typed_config:
          "@type": type.http.filters.RateLimit
          domain: "ecommerce-ratelimit"  # 域名，对应Rate Limit Server配置
          stage: "production"
          failure_mode_deny: false       # Rate Limit Server不可达时不拒绝
          rate_limit_service:
            grpc_service:
              envoy_grpc:
                cluster_name: rate_limit_cluster  # 外部gRPC服务
                authority: ratelimit.example.com
            transport_api_version: V3
          timeout:
            seconds: 1
          rate_limits:
          - unit_name: "per_minute"       # 限流维度名称
            actions:
            - { generic_key: { descriptor_value: "orders" } }  # 匹配所有订单相关请求

/**
 * 自适应限流器实现
 * 
 * 思路：
 * 1. 监控最近N个请求的响应时间
 * 2. 计算P99延迟
 * 3. 如果P99 > 阈值，降低限流额度（类似TCP的Multiplicative Decrease）
 * 4. 如果P99 < 阈值，缓慢增加限流额度（Additive Increase）
 */
@Component
public class AdaptiveRateLimiter {
 
    private final ConcurrentMap<String, AdaptiveState> stateMap = new ConcurrentHashMap<>();
    
    private static final double INCREASE_FACTOR = 1.05;   # 增加5%（AI）
    private static final double DECREASE_FACTOR = 0.8;    # 减少20%（MD）
    private static final long MIN_QPS = 10;               # 最小QPS下限
    private static final long MAX_QPS = 10000;             # 最大QPS上限
    private static final long LATENCY_THRESHOLD_MS = 500;  # P99延迟阈值
    
    /**
     * 检查是否允许通过（自适应限流）
     */
    public synchronized AdaptiveResult checkAdaptive(String serviceKey) {
        AdaptiveState state = stateMap.computeIfAbsent(serviceKey, k -> 
            new AdaptiveState(MAX_QPS / 2)  # 初始值为最大值的一半
        );
 
        long now = System.currentTimeMillis();
        
        # 记录本次请求
        state.recordRequest(now);
        
        # 检查是否超过当前限额
        if (state.getCurrentQps() > state.getAllowedQps()) {
            return AdaptiveResult.rejected(state.getAllowedQps(), calculateRetryAfter(state));
        }
 
        # 放行
        state.incrementAllowed();
        return AdaptiveResult.allowed(state.getAllowedQps());
    }
 
    /**
     * 根据响应时间调整限流额度（应该在请求完成后调用）
     */
    public synchronized void recordLatency(String serviceKey, long latencyMs) {
        AdaptiveState state = stateMap.get(serviceKey);
        if (state == null) return;
 
        state.recordLatency(latencyMs);
 
        // 计算最近的P99延迟
        long p99 = state.getP99Latency();
 
        if (p99 > LATENCY_THRESHOLD_MS) {
            # 响应太慢：减少额度（乘法递减）
            long newAllowed = Math.max(MIN_QPS, (long)(state.getAllowedQps() * DECREASE_FACTOR));
            state.setAllowedQps(newAllowed);
            
            log.info("[AdaptiveRateLimit] Decreasing QPS for {}: {} -> {} (p99={}ms)",
                     serviceKey, state.getAllowedQps(), newAllowed, p99);
        } else {
            # 响应正常：缓慢增加额度（线性增长）
            long newAllowed = Math.min(MAX_QPS, (long)(state.getAllowedQps() * INCREASE_FACTOR));
            state.setAllowedQps(newAllowed);
            
            if (state.getRequestCount() % 100 == 0) {  # 每100次打印一次
                log.debug("[AdaptiveRateLimit] Increasing QPS for {}: {} -> {} (p99={}ms)",
                         serviceKey, state.getAllowedQps(), newAllowed, p99);
            }
        }
    }
 
    @Data
    private static class AdaptiveState {
        private volatile long allowedQps;
        private final LongAdder requestCounter = new LongAdder();
        private final LongAdder allowedCounter = new LongAdder();
        private final CircularArray latencyHistory = new CircularArray(1000);  // 最近1000个请求的延迟
        private long lastAdjustTime;
        
        // ... 实现略 ...
    }
}

# nginx.conf - 接入层限流
limit_req_zone $binary_remote_addr zone=ip_limit:10m rate=100r/s;  # 每IP每秒100次
limit_req_zone $server_name zone=global_limit:10m rate=10000r/s;  # 全局每秒10000次
 
server {
    location /api/ {
        limit_req zone=global_limit burst=200 nodelay;  # 允许突发200
        limit_req zone=ip_limit burst=20 nodelay;         # 每IP突发20
        
        limit_req_status 429;  # 被限流时返回429
        
        proxy_pass http://backend;
    }
}